import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

public class RAJoin implements RAExpression, Cloneable {

	// These left and right subexpressions are only needed if the join has a where clause with
	// predicates that can be pushed down or additional RA operations that must occur on the
	// left or right of the join before the join.
	private RAExpression left = null;
	private RAExpression right = null;

	private Map<String, TableData> catalog;
	private boolean evaluateSelectList = false;
	private ArrayList<Expression> select = null;
//	private ArrayList<Expression> columnsToProject = null;
	private Map<String, String> from;
	private Expression where;
	
	// Array list of selection predicates which all must be true
	// in order for a record to be selected.
	// JES TODO: This field should be turned into a local variable where it is used.
	private ArrayList<Expression> whereDisjunctions = null;
	
	private ArrayList<Expression> leftIntermediateSelect;
	private ArrayList<Expression> rightIntermediateSelect;
	
	public RAJoin(Map<String, TableData> catalog, RAExpression left, RAExpression right, Map<String, String> from)
	{
		this.left = left;
		this.right = right;
		
		this.catalog = catalog;
		this.evaluateSelectList = false;
		this.select = null;
//		this.columnsToProject = null;
		
		this.from = from;
		this.where = null;
	}

	/**
	 * Not a great implementation of clone.
	 */
	protected RAJoin clone()
	{
		try
		{
			return (RAJoin) super.clone();
		}
		catch (CloneNotSupportedException e)
		{
			throw new RuntimeException(e);
		}
	}

	public RAExpression getLeft()
	{
		return left;
	}

	public void setLeft(final RAExpression left)
	{
		this.left = left;
	}

	public RAExpression getRight()
	{
		return right;
	}

	public void setRight(final RAExpression right)
	{
		this.right = right;
	}

	@Override
	public RAOutput execute() throws IOException
	{
		if(DBConfig.DEBUG)
		{
			System.out.println("Executing RAJoin: " + this.toString());
		}
		// ArrayList <Attribute> inAttsRight = new ArrayList <Attribute> ();
		// inAttsRight.add (new Attribute ("Int", "o_orderkey"));
		// inAttsRight.add (new Attribute ("Int", "o_custkey"));
		// inAttsRight.add (new Attribute ("Str", "o_orderstatus"));
		// inAttsRight.add (new Attribute ("Float", "o_totalprice"));
		// inAttsRight.add (new Attribute ("Str", "o_orderdate"));
		// inAttsRight.add (new Attribute ("Str", "o_orderpriority"));
		// inAttsRight.add (new Attribute ("Str", "o_clerk"));
		// inAttsRight.add (new Attribute ("Int", "o_shippriority"));
		// inAttsRight.add (new Attribute ("Str", "o_comment"));

		// This code is suboptimal because the in attributes in general
		// will not be the same as the out attributes. The code to
		// RASelection needs to be improved.
		// RAOutput rightOut = right.execute();
		// ArrayList<Attribute> inAttsRight = rightOut.inAtts;
		RAOutput rightOut = null;
		ArrayList<Attribute> inAttsRight = null;
		String rightInFile = null;
		if(right != null)
		{
			rightOut = right.execute();
			inAttsRight = rightOut.outAtts;
			rightInFile = rightOut.outFile;
		}
		else
		{
			throw new RuntimeException("RAJoin invalid: right == null && rightFrom == null");
		}

		// ArrayList <Attribute> inAttsLeft = new ArrayList <Attribute> ();
		// inAttsLeft.add (new Attribute ("Int", "c_custkey"));
		// inAttsLeft.add (new Attribute ("Str", "c_name"));
		// inAttsLeft.add (new Attribute ("Str", "c_address"));
		// inAttsLeft.add (new Attribute ("Int", "c_nationkey"));
		// inAttsLeft.add (new Attribute ("Str", "c_phone"));
		// inAttsLeft.add (new Attribute ("Float", "c_acctbal"));
		// inAttsLeft.add (new Attribute ("Str", "c_mktsegment"));
		// inAttsLeft.add (new Attribute ("Str", "c_comment"));

		// This code is suboptimal because the in attributes in general
		// will not be the same as the out attributes. The code to
		// RASelection needs to be improved.
		// RAOutput leftOut = left.execute();
		// ArrayList<Attribute> inAttsLeft = leftOut.outAtts;
		RAOutput leftOut = null;
		ArrayList<Attribute> inAttsLeft = null;
		String leftInFile = null;
		if(left != null)
		{
			leftOut = left.execute();
			inAttsLeft = leftOut.outAtts;
			leftInFile = leftOut.outFile;
		}

		// ArrayList <Attribute> outAtts = new ArrayList <Attribute> ();
		// outAtts.add (new Attribute ("Str", "att1"));
		// outAtts.add (new Attribute ("Int", "att2"));
		// outAtts.add (new Attribute ("Int", "att3"));
		// outAtts.add (new Attribute ("Str", "att4"));
		// outAtts.add (new Attribute ("Int", "att5"));

		ArrayList<Attribute> outAtts = null;
		int leftOutSize = 0;
		int rightOutSize = 0;
//		if(select == null)
		if(!evaluateSelectList)
		{
			outAtts = new ArrayList<Attribute>();
			
			if(leftIntermediateSelect != null) // leftFrom != null
			{
				Map<String, String> leftTables = new HashMap<String, String>();
				left.getTables(leftTables);
				// leftIntermediateSelect should be only identifiers at this point
				leftOutSize = leftIntermediateSelect.size();
				for(int i = 0; i < leftOutSize; i++)
				{
					Expression e = leftIntermediateSelect.get(i);
					outAtts.add(new Attribute(e.getReturnType(leftTables, catalog), "att" + i));
				}
			}
			else if(left != null)
			{
				leftOutSize = inAttsLeft.size();
				for(int i = 0; i < leftOutSize; i++)
				{
					Attribute a = inAttsLeft.get(i);
					outAtts.add(new Attribute(a.getType(), "att" + i));
				}
			}
			else
			{
				// This table isn't referenced in the output, so don't do anything with it.
			}

			if(rightIntermediateSelect != null) // rightFrom != null
			{
				Map<String, String> rightTables = new HashMap<String, String>();
				right.getTables(rightTables);
				rightOutSize = rightIntermediateSelect.size();
				for(int i = 0; i < rightOutSize; i++)
				{
					Expression e = rightIntermediateSelect.get(i);
					outAtts.add(new Attribute(e.getReturnType(rightTables, catalog), "att" + (leftOutSize + i)));
				}
			}
			else if(right != null)
			{
				rightOutSize = inAttsRight.size();
				for(int i = 0; i < rightOutSize; i++)
				{
					Attribute a = inAttsRight.get(i);
					outAtts.add(new Attribute(a.getType(), "att" + (leftOutSize + i)));
				}
			}
			else
			{
				// This table isn't referenced in the output, so don't do anything with it.
			}
			
		}
		else
		{
			int selectSize = select.size();
			outAtts = new ArrayList<Attribute>(selectSize);
			for(int i = 0; i < selectSize; i++)
			{
				outAtts.add(new Attribute(select.get(i).getReturnType(from, catalog), "att" + i));
			}
		}

		final ArrayList<Expression> joinEqualityPredicates = new ArrayList<Expression>();
		if(where != null)
		{
			whereDisjunctions = where.toDisjunctions();
			for(final Iterator<Expression> it = whereDisjunctions.iterator(); it.hasNext();)
			{
				final Expression e = it.next();
				if(e.getType().equals("equals"))
				{
					final Expression l = e.getSubexpression("left");
					final Expression r = e.getSubexpression("right");
					if(!l.getType().equals("identifier") || !r.getType().equals("identifier"))
					{
						continue;
					}
					final String joinString = e.toJoinString(inAttsLeft, inAttsRight);
					if (joinString.contains("left.") && joinString.contains("right."))
					{
						joinEqualityPredicates.add(e);
						it.remove();
					}
				}
			}
		}
		// Do not rejunct. The above essentially is setting up hashing columns. The where clause still
		// needs to have the equality predicate to check the hashed values.
		// where = Expression.reJunct(whereDisjunctions);

		// ArrayList <String> leftHash = new ArrayList <String> ();
		// leftHash.add ("c_custkey");

		// Need to enhance, otherwise this is a nested loop cross join
		final ArrayList<String> leftHash = new ArrayList<String>();

		// ArrayList <String> rightHash = new ArrayList <String> ();
		// rightHash.add ("o_custkey");

		// Need to enhance, otherwise this is a nested loop cross join
		final ArrayList<String> rightHash = new ArrayList<String>();

		// Get the joinEqualityPredicates into the hash lists
		outerFor:
		for(final Expression e : joinEqualityPredicates)
		{
			final String eLeft = e.getSubexpression("left").getValue().replace('.', '_');
			final String eRight = e.getSubexpression("right").getValue().replace('.', '_');
			if (DBConfig.DEBUG)
			{
				System.out.println("Considering expression " + eLeft + " = " + eRight + ".");
			}
			for(final Attribute a : inAttsLeft)
			{
				if(a.getName().equals(eLeft))
				{
					if (DBConfig.DEBUG)
					{
						System.out.println("adding " + eLeft + " to leftHash; adding " + eRight	+ " to rightHash.");
					}
					leftHash.add(eLeft);
					rightHash.add(eRight);
					continue outerFor;
				}
			}
			if(DBConfig.DEBUG)
			{
				System.out.println("adding " + eRight + " to leftHash; adding "	+ eLeft + " to rightHash.");
			}
			leftHash.add(eRight);
			rightHash.add(eLeft);
		}

		// String selection =
		// "right.o_custkey == left.c_custkey && right.o_custkey > Int (1000)";

		String selection = null;
		if(where != null)
		{
			// toJoinString replaces the '.' between the correlation and the identifier name with
			// an underscore and prefixes the new identifier name with a "left." or "right.".
			selection = where.toJoinString(inAttsLeft, inAttsRight);
		}
		else
		{
			selection = "true";
		}
		if (DBConfig.DEBUG)
		{
			System.out.println("Joining with selection string " + selection + ".");
		}

		// HashMap <String, String> exprs = new HashMap <String, String> ();
		// exprs.put ("att1", "right.o_comment + Str(\" \") + left.c_comment");
		// exprs.put ("att2", "right.o_custkey");
		// exprs.put ("att3", "left.c_custkey");
		// exprs.put ("att4", "left.c_name");
		// exprs.put ("att5", "right.o_orderkey");

		final HashMap<String, String> exprs = new HashMap<String, String>();
//		if(select != null)
		if(evaluateSelectList)
		{
			// This is a top level select, and we just need to map everything to the final expressions.
			for (int i = 0; i < select.size(); i++)
			{
				exprs.put("att" + i, select.get(i).toJoinString(inAttsLeft, inAttsRight));
			}
		}
		else
		{
			// This is not a top level select and the only things we are selecting are attributes.
			// We are possibly in future enhancements going to project away attributes here or do
			// early evaluation of more complex expressions.
			int outAttsSize = outAtts.size();
			for(int i = 0; i < outAttsSize; i++)
			{
				String attName = "att" + i;
				if(i < leftOutSize)
				{
					int leftIndex = i;
//					if(left != null)
//					{
//						exprs.put(attName, "left." + inAttsLeft.get(leftIndex).getName());
//					}
//					else
						if(this.leftIntermediateSelect != null)
					{
						exprs.put(attName, "left." + leftIntermediateSelect.get(leftIndex).toExecutionString());
					}
					else
					{
//						throw new RuntimeException("RAJoin Execution Error: left == null && leftIntermediateSelect == null");
					}
				}
				else
				{
					int rightIndex = i - leftOutSize;
//					if(right != null)
//					{
//						exprs.put(attName, "right." + inAttsRight.get(rightIndex).getName());
//					}
//					else
						if(this.rightIntermediateSelect != null)
					{
						exprs.put(attName, "right." + rightIntermediateSelect.get(rightIndex).toExecutionString());
					}
					else
					{
//						throw new RuntimeException("RAJoin Execution Error: right == null && rightIntermediateSelect == null");
					}
				}
			}
		}

		// run the join
		/*
		 * try { Join foo = new Join (inAttsLeft, inAttsRight, outAtts,
		 * leftHash, rightHash, selection, exprs, "customer.tbl", "orders.tbl",
		 * "out.tbl", "g++", "cppDir/"); } catch (Exception e) { throw new
		 * RuntimeException (e); }
		 */

		// String leftInFile = leftOut.outFile;
		// String rightInFile = rightOut.outFile;
		final String outFile = leftInFile.replaceAll("\\.[sj]out", "").replaceAll("\\.tbl", "")
				+ "." + rightInFile.replaceAll("\\.[sj]out", "").replaceAll("\\.tbl", "") + ".jout";

		if(DBConfig.DEBUG)
		{
			System.out.println("inAttsLeft: " + inAttsLeft.toString());
			System.out.println("inAttsRight: " + inAttsRight.toString());
			System.out.println("outAtts: " + outAtts.toString());
			System.out.println("leftHash: " + leftHash.toString());
			System.out.println("rightHash: " + rightHash.toString());
			System.out.println("selection: " + selection);
			System.out.println("exprs: " + exprs.toString());
		}
		
		try
		{
			// new Join (inAttsLeft, inAttsRight, outAtts, leftHash, rightHash,
			// selection, exprs,
			// leftOut.outFile, rightOut.outFile, outFile,
			// Interpreter.cppCompiler, Interpreter.cppTmpDir);
			new Join(inAttsLeft, inAttsRight, outAtts, leftHash, rightHash,
					selection, exprs, leftInFile, rightInFile, outFile,
					DBConfig.CPP_COMPILER, DBConfig.CPP_TEMP_DIR);
		}
		catch(final Exception e)
		{
			throw new RuntimeException(e);
		}

		// This has been moved to Join
//		if(left != null && !leftInFile.endsWith(".tbl"))
//		{
//			final File f = new File(leftInFile);
//			f.delete();
//		}
//
//		if(right != null && !rightInFile.endsWith(".tbl"))
//		{
//			final File f = new File(rightInFile);
//			f.delete();
//		}

		// This will have to change if I try projecting away evaluated expressions early on
		ArrayList<Attribute> outAttributes = new ArrayList<Attribute>(outAtts.size());
		outAttributes.addAll(inAttsLeft);
		outAttributes.addAll(inAttsRight);
		
		// Need to get rid of inAttributes and select from RAOutput data structure
		return new RAOutput(outFile, /*outAttributes,*/ outAttributes/*, select*/);
	}

	@Override
	public <T> T accept(final RAExpressionVisitor<T> v, final Object... args)
	{
		return v.joinCase(this, args);
	}

	@Override
	public RAStatistics estimateCost(Map<String, TableData> catalog,
			Map<String, TableData> tempCatalog)
	{
		if(DBConfig.DEBUG)
		{
			validateState();
		}
		// I need to change this check since it is actually not necessarily a valid error.
		// RESUME HERE and remove this check.
		
		long leftCard = 0;
		long rightCard = 0;
		
		long leftCost = 0;
		long rightCost = 0;

		Map<String, TableData> leftTempCatalog = new HashMap<String,TableData>();
		Map<String, TableData> rightTempCatalog = new HashMap<String, TableData>();
		
		if(left != null)
		{
			RAStatistics leftStats = left.estimateCost(catalog, leftTempCatalog);
			leftCard = leftStats.getCardinality();
			leftCost = leftStats.getCost();
		}
		else
		{
			throw new RuntimeException("RAJoin invalid: leftFrom == null && left == null");
		}
		
		if(right != null)
		{
			RAStatistics rightStats = right.estimateCost(catalog, rightTempCatalog);
			rightCard = rightStats.getCardinality();
			rightCost = rightStats.getCost();
		}
		else
		{
			throw new RuntimeException("RAJoin invalid: rightFrom == null && right == null");
		}
		
		long cost = leftCost + rightCost;
		
		if(where == null)
		{
			long card = leftCard * rightCard;
			cost += card;
			DBUtils.mergeCatalogs(tempCatalog, leftTempCatalog);
			DBUtils.mergeCatalogs(tempCatalog, rightTempCatalog);
			DBUtils.setCatalogMaxCard(tempCatalog, card);
			return new RAStatistics(card, cost);
		}
		
		// Get the complete from list for the left side and right side
		Map<String, String> leftTables;
		Map<String, String> rightTables;
		
		if(left != null)
		{
			leftTables = new HashMap<String, String>();
			left.getTables(leftTables);
		}
		else
		{
			throw new RuntimeException("RAJoin invalid state: leftFrom == null && left == null");
		}
		
		if(right != null)
		{
			rightTables = new HashMap<String, String>();
			right.getTables(rightTables);
		}
		else
		{
			throw new RuntimeException("RAJoin invalid state: right == null && rightFrom == null");
		}
		
		// Join filter factor is of the form
		// join_card_multiplier * left_filter_fact * right_filter_factor
		// where join_card_multiplier is the minimum cardinality of either table in any of the
		// fields being joined.
		double joinCardMultiplier = Double.MAX_VALUE;
		double leftFilterFactor = 1.0;
		double rightFilterFactor = 1.0;
		whereDisjunctions = where.toDisjunctions();
		Map<String, TableData> updatedTempCatalog = new HashMap<String, TableData>(); // JES - THIS MAY BE REDUNDANT
		for(Expression e : whereDisjunctions)
		{
			// Assuming that the where clause is a valid where clause, we should now be able to
			// estimate the cardinality
			double tmpLeftFilterFactor = e.computeFilterFactor(leftTables, catalog, leftTempCatalog, updatedTempCatalog);
			leftFilterFactor *= tmpLeftFilterFactor;
			
			double tmpRightFilterFactor = e.computeFilterFactor(rightTables, catalog, rightTempCatalog, updatedTempCatalog);
			rightFilterFactor *= tmpRightFilterFactor;
			
			if(e.getType().equals("equals"))
			{
				joinCardMultiplier = Math.min(joinCardMultiplier, 1.0 / tmpLeftFilterFactor);
				joinCardMultiplier = Math.min(joinCardMultiplier, 1.0 / tmpRightFilterFactor);
			}
			else if(e.getType().equals("not"))
			{
				joinCardMultiplier = Math.min(joinCardMultiplier, leftCard * tmpLeftFilterFactor);
				joinCardMultiplier = Math.min(joinCardMultiplier, rightCard * tmpRightFilterFactor);
			}
		}
		
		double minLeftFilterFactor = 1.0 / leftCard;
		if(leftFilterFactor < minLeftFilterFactor)
		{
			leftFilterFactor = minLeftFilterFactor;
		}
		
		double minRightFilterFactor = 1.0 / rightCard;
		if(rightFilterFactor < minRightFilterFactor)
		{
			rightFilterFactor = minRightFilterFactor;
		}
		
		if(joinCardMultiplier > leftCard || joinCardMultiplier > rightCard)
		{
			joinCardMultiplier = Math.min(leftCard, rightCard);
		}
		
		double joinFilterFactor = joinCardMultiplier * leftFilterFactor * rightFilterFactor;
		long card = Math.round(joinFilterFactor * leftCard * rightCard);
		
		// I need to somehow combine the leftTempCatalog and the rightTempCatalog into the tempCatalog that
		// was passed as input to this method. Assuming that the table names are unique in each, it is a simple
		// task of combining. Assuming table names are non-unique, then a special merge must take place in which
		// the minimum of each attribute's cardinality must be used instead.
		
		// This behavior should be encapsulated into a class method. Ideally the concept of a catalog needs
		// to become a class rather than a simple data structure like a hash map.
		tempCatalog.putAll(leftTempCatalog);
		
		DBUtils.mergeCatalogs(tempCatalog, rightTempCatalog);
		DBUtils.mergeCatalogs(tempCatalog, updatedTempCatalog);
		
		DBUtils.setCatalogMaxCard(tempCatalog, card);
		
		return new RAStatistics(card, cost + card);
	}

	@Override
	public void getTables(Map<String, String> tables)
	{
		if(left != null)
		{
			left.getTables(tables);
		}
		
		if(right != null)
		{
			right.getTables(tables);
		}
	}
	

	@Override
	public ArrayList<RAExpression> getJoinTransformations()
	{
		if((!(left instanceof RAJoin)) && (!(right instanceof RAJoin)))
		{
			return null;
		}
		
		ArrayList<RAExpression> transformations = new ArrayList<RAExpression>();
		
		// (A JOIN (B JOIN C)) -> (C JOIN (B JOIN A)) and (B JOIN (A JOIN C))
		if(right instanceof RAJoin)
		{
			RAJoin rightJoin = (RAJoin) right;
			
			// (A JOIN (B JOIN C)) -> (B JOIN (A JOIN C))
			RAJoin rightClone1 = rightJoin.clone();
			// (A JOIN (B JOIN C)) -> (C JOIN (B JOIN A))
			RAJoin rightClone2 = rightJoin.clone();
			
			RAJoin clone1 = this.clone();
			clone1.setRight(rightClone1);
			//clone1.right = rightClone1;

			RAJoin clone2 = this.clone();
			clone2.setRight(rightClone2);
			//clone2.right = rightClone2;
			
			transformations.add(clone1);
			transformations.add(clone2);
			
			if(left != null)
			{
				if(rightJoin.left != null)
				{
					clone1.left = rightJoin.left;
					rightClone1.left = left;
				}
				else
				{
					throw new RuntimeException("RAJoin structure invalid since rightJoin.left == null && rightJoin.leftFrom == null");
				}
				
				if(rightJoin.right != null)
				{
					clone2.left = rightClone2.right;
					rightClone2.right = left;
				}
				else
				{
					throw new RuntimeException("RAJoin structure invalid since rightJoin.right == null && rightJoin.rightFrom == null");
				}
			}
			else
			{
				throw new RuntimeException("RAJoin structure invalid since left == null && leftFrom == null");
			}

			// transfer the predicates
			if(rightClone1.where != null)
			{
				clone1.addWhereDisjunction(rightClone1.where);
				rightClone1.where = null;
			}
			clone1.pushDownSelections();
			
			if(rightClone2.where != null)
			{
				clone2.addWhereDisjunction(rightClone2.where);
				rightClone2.where = null;
			}
			clone2.pushDownSelections();
		}
		
		if(left instanceof RAJoin)
		{
			RAJoin leftJoin = (RAJoin) left;

			// ((A JOIN B) JOIN C) -> ((C JOIN B) JOIN A)
			RAJoin leftClone1 = leftJoin.clone();
			// ((A JOIN B) JOIN C)) -> ((A JOIN C) JOIN B)
			RAJoin leftClone2 = leftJoin.clone();
			
			RAJoin clone1 = this.clone();
			clone1.left = leftClone1;

			RAJoin clone2 = this.clone();
			clone2.left = leftClone2;
			
			transformations.add(clone1);
			transformations.add(clone2);
			
			if(right != null)
			{
				if(leftJoin.left != null)
				{
					clone1.right = leftJoin.left;
					leftClone1.left = right;
				}
				else
				{
					throw new RuntimeException("RAJoin structure invalid since rightJoin.left == null && rightJoin.leftFrom == null");
				}
				
				if(leftJoin.right != null)
				{
					clone2.right = leftClone2.right;
					leftClone2.right = right;
				}
				else
				{
					throw new RuntimeException("RAJoin structure invalid since rightJoin.right == null && rightJoin.rightFrom == null");
				}
			}
			else
			{
				throw new RuntimeException("RAJoin structure invalid since left == null && leftFrom == null");
			}

			// transfer the predicates
			if(leftClone1.where != null)
			{
				clone1.addWhereDisjunction(leftClone1.where);
				leftClone1.where = null;
			}
			clone1.pushDownSelections();
			
			if(leftClone2.where != null)
			{
				clone2.addWhereDisjunction(leftClone2.where);
				leftClone2.where = null;
			}
			clone2.pushDownSelections();
		}
		
		if(left instanceof RAJoin)
		{
			ArrayList<RAExpression> leftTransformations = left.getJoinTransformations();
			if(leftTransformations != null)
			{
				for(RAExpression ra : leftTransformations)
				{
					RAJoin clone = this.clone();
					clone.left = ra;
					transformations.add(clone);
					if(DBConfig.DEBUG)
					{
						clone.validateState();
					}
				}
			}
		}
		
		if(right instanceof RAJoin)
		{
			ArrayList<RAExpression> rightTransformations = right.getJoinTransformations();
			if(rightTransformations != null)
			{
				for(RAExpression ra : rightTransformations)
				{
					RAJoin clone = this.clone();
					clone.right = ra;
					transformations.add(clone);
					if(DBConfig.DEBUG)
					{
						clone.validateState();
					}
				}
			}
		}
		
		if(transformations != null && transformations.size() > 0)
		{
			return transformations;
		}
		else
		{
			return null;
		}
	}

	/**
	 * Pushes down selection predicates one level if possible. This should perhaps be modified
	 * to do recursive push down. That would probably simplify the design.
	 * JES TO DO TODO: this should be enhanced or replaced by the newer version of this method that I wrote, pushdownJoinPredicates
	 */
	private void pushDownSelections()
	{
		if(where == null)
		{
			return;
		}
		
		whereDisjunctions = where.toDisjunctions();
		for(Iterator<Expression> it = whereDisjunctions.iterator(); it.hasNext();)
		{
			Expression e = it.next();
			boolean referencesLeft = leftReferencesExpression(e);
			boolean referencesRight = rightReferencesExpression(e);
			
			if(referencesLeft && !referencesRight)
			{
				if(left != null)
				{
					it.remove();
					left.addWhereDisjunction(e);
				}
			}
			else if(!referencesLeft && referencesRight)
			{
				if(right != null)
				{
					it.remove();
					right.addWhereDisjunction(e);
				}
			}
		}
		where = Expression.reJunct(whereDisjunctions);
	}
	
	private boolean leftReferencesExpression(Expression e)
	{
		if(left != null)
		{
			return left.referencedByExpression(e);
		}
		else
		{
			throw new RuntimeException("RAJoin invalid: leftFrom == null && left == null");
		}
	}
	
	private boolean rightReferencesExpression(Expression e)
	{
		if(right != null)
		{
			return right.referencedByExpression(e);
		}
		else
		{
			throw new RuntimeException("RAJoin invalid: rightFrom == null && right == null");
		}
	}

	public void addWhereDisjunction(Expression e)
	{
		if(where == null)
		{
			where = e;
		}
		else if(e != null)
		{
			Expression newWhere = new Expression("and");
			newWhere.setSubexpression(where, e);
			where = newWhere;
		}
	}

	@Override
	public boolean referencedByExpression(Expression e)
	{
		if(left != null)
		{
			if(left.referencedByExpression(e))
			{
				return true;
			}
		}
		
		if(right != null)
		{
			if(right.referencedByExpression(e))
			{
				return true;
			}
		}
		
		return false;
	}

	@Override
	public String joinPlanToString()
	{
		String leftJoinPlan = "";
		String rightJoinPlan = "";
		if(left != null)
		{
			leftJoinPlan = left.joinPlanToString();
		}
		else
		{
			throw new RuntimeException("RAJoin invalid structure: leftFrom == null && left == null");
		}
		
		if(right != null)
		{
			rightJoinPlan = right.joinPlanToString();
		}
		else
		{
			throw new RuntimeException("RAJoin invalid structure: rightFrom == null && right == null");
		}
		
		return "(" + leftJoinPlan + " JOIN " + rightJoinPlan + ")";
	}
	
	public String toString()
	{
		return this.joinPlanToString();
	}
	
	public void validateState()
	{
		if((left == null) || (right == null))
		{
			System.err.println("RAJoin Invalid state.");
		}
	}

	@Override
	public void pushdownReferencedColumnsList(ArrayList<Expression> selectList)
	{
		if(left != null)
		{
			left.pushdownReferencedColumnsList(selectList);
		}
		if(right != null)
		{
			right.pushdownReferencedColumnsList(selectList);
		}
		
		leftIntermediateSelect = new ArrayList<Expression>();
		rightIntermediateSelect = new ArrayList<Expression>();
		for(Expression e : selectList)
		{
			if(left != null && left.referencedByExpression(e))
			{
				leftIntermediateSelect.add(e);
			}
			else if(right != null && right.referencedByExpression(e))
			{
				rightIntermediateSelect.add(e);
			}
		}
	}

	@Override
	public void applySelectList(ArrayList<Expression> select)
	{
		this.evaluateSelectList = true;
		this.select = select;
	}

	@Override
	public void pushdownJoinPredicates(ArrayList<Expression> joinWhereClauses)
	{
		for(Iterator<Expression> it = joinWhereClauses.iterator(); it.hasNext();)
		{
			Expression e = it.next();
			
			boolean referencesLeft = false;
			if(left != null)
			{
				if(left.referencedByExpression(e))
				{
					referencesLeft = true;
				}
				else
				{
					continue;
				}
			}

			boolean referencesRight = false;
			if(right != null)
			{
				if(right.referencedByExpression(e))
				{
					referencesRight = true;
				}
				else
				{
					continue;
				}
			}
			
			if(referencesLeft && referencesRight)
			{
				it.remove();
				this.addWhereDisjunction(e);
			}
		}
		
		if(left != null)
		{
			left.pushdownJoinPredicates(joinWhereClauses);
		}
		
		if(right != null)
		{
			right.pushdownJoinPredicates(joinWhereClauses);
		}
	}
}
